//package edu.csl.study.spark.kafka.v08
//
//import kafka.serializer.StringDecoder
//import org.apache.log4j.{Level, Logger}
//import org.apache.spark.SparkConf
//import org.apache.spark.storage.StorageLevel
//import org.apache.spark.streaming.kafka.KafkaUtils
//import org.apache.spark.streaming.{Seconds, StreamingContext}
//
///**
// * Kafka 0.8
// * 方式一：基于Reciver方式。 实际工作中不用这种方式
// *   如何保证数据不丢失：开启WAL机制，但是这样会导致性能差。
// *  <dependency>
// *    <groupId>org.apache.spark</groupId>
// *    <artifactId>spark-streaming-kafka-0-8_2.11</artifactId>
// *    <version>${apache.spark.version}</version>
// *  </dependency>
// */
//object ReceiverKafkaWordCount {
//  def main(args: Array[String]): Unit = {
//    Logger.getLogger("org").setLevel(Level.ERROR)
//    //步骤一：初始化程序入口
//    val sparkConf = new SparkConf().setMaster("local[2]").setAppName("ReceiverKafkaWordCount")
//    val ssc = new StreamingContext(sparkConf, Seconds(5))
//
//    val kafkaParams =  Map[String, String](
//      "zookeeper.connect"->"centos20:2181,centos21:2181,centos22:2181",
//      "group.id" -> "testSpark"
//    )
//    //可以多个主题
//    val topics = "flink,spark,hive".split(",").map((_,1)).toMap
//    //步骤二：获取数据源
//    //默认只会有一个receiver
//
//    val lines = KafkaUtils.createStream[String,String,StringDecoder,StringDecoder](
//     ssc,kafkaParams,topics, StorageLevel.MEMORY_AND_DISK_SER)
//    //多个receiver：3个receiver
////    val kafkaStreams = (1 to 3).map(_ => {
////  KafkaUtils.createStream[String, String, StringDecoder, StringDecoder](
////    ssc, kafkaParams, topics, StorageLevel.MEMORY_AND_DISK_SER)
////})
//      //调用union，3个receiver合并起来。
////    val lines = ssc.union(kafkaStreams)
//
//    //步骤三：业务代码处理
//    lines.map(_._2).flatMap(_.split(",")).map((_,1)).reduceByKey(_+_).print()
//    ssc.start()
//    ssc.awaitTermination()
//    ssc.stop()
//  }
//
//}
